package com.xiashitech.apiservice.grpc;

import cn.hutool.json.JSONUtil;
import com.xiashitech.apiservice.dto.AuthFactDTO;
import com.xiashitech.apiservice.service.IAuthFactService;
import com.xiashitech.apiservice.util.BeanCopierUtil;
import com.xiashitech.interfaces.agent.dto.SystemInfoDTO;
import com.xiashitech.interfaces.apiservice.dto.base.Table;
import com.xiashitech.interfaces.apiservice.dto.AuthFactRequest;
import com.xiashitech.interfaces.apiservice.dto.AuthFactResponse;
import com.xiashitech.interfaces.grpc.RPCAuthFactRequest;
import com.xiashitech.interfaces.grpc.RPCAuthFactResponse;
import com.xiashitech.interfaces.grpc.RPCAuthFactServiceGrpc;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Slf4j
//@Service
@GrpcService
public class AuthFactService extends RPCAuthFactServiceGrpc.RPCAuthFactServiceImplBase {

    // 一个 Agent 维持一个
    private final ConcurrentHashMap<String, StreamObserver<RPCAuthFactResponse>> responseAuthFactMap = new ConcurrentHashMap<>();

    private final ConcurrentHashMap<String, AuthFactRequest> requestMap = new ConcurrentHashMap<>();
    @Autowired
    private IAuthFactService authFactService;


    @Override
    public StreamObserver<RPCAuthFactRequest> biStreamAuthFact(StreamObserver<RPCAuthFactResponse> responseObserver) {
        // return super.biStreamAuthFact(responseObserver);
        log.info("biStreamAuthFact enter...........");
        StreamObserver<RPCAuthFactRequest> requestStreamObserver = new StreamObserver<RPCAuthFactRequest>() {
            @Override
            public void onNext(RPCAuthFactRequest value) {
                try {
                    log.info(" received the message: response identityhashcode: " + System.identityHashCode(responseObserver));
                    String requestKey = value.getRequestKey();
                    if(requestKey == null || "".equals(requestKey)) {
                        log.info("request info is empty, noneed response. ");
                        return;
                    }

                    // 对收到的 request 进行处理，获取到 response 对象，
                    AuthFactRequest authFactRequest = JSONUtil.toBean(value.getAuthFactRequestJson(), AuthFactRequest.class);
                    String requestKey1 =  String.valueOf(authFactRequest.hashCode()); // 应该requestKey1 与 requestKey 是一样的
                    requestMap.put(requestKey1, authFactRequest);

                    // 不同的 requestKey1 会很多，Map 会有很多相同的 responseObserver
                    responseAuthFactMap.put(requestKey1, responseObserver);

                    AuthFactDTO authFactDTO = authFactService.supplementUserTracing(AuthFactDTO.builder().systemId(authFactRequest.getSystemId()).userId(authFactRequest.getUserId()).sign(authFactRequest.getSign()).appCode(authFactRequest.getAppCode())
                            .instance(authFactRequest.getDb().getInstance()).database(authFactRequest.getDb().getDatabase()).schema(authFactRequest.getDb().getSchema())
                            .sql(authFactRequest.getDb().getSql()).ip(authFactRequest.getDb().getIp()).port(authFactRequest.getDb().getPort()).type(authFactRequest.getDb().getType()).sourceSystemId(authFactRequest.getSourceSystemId()).apiName(authFactRequest.getApiName()).apiSystemId(authFactRequest.getApiSystemId()).build());
                    AuthFactResponse authFactResponse = BeanCopierUtil.copy(authFactDTO, AuthFactResponse.class);
                    log.info("AuthFactDTO: "+JSONUtil.toJsonStr(authFactDTO));
                    log.info("AuthFactResponse: "+JSONUtil.toJsonStr(authFactResponse));
                    // TODO 没取到 authFactDTO 的处理 ？
                    RPCAuthFactResponse rpcResponseObject = RPCAuthFactResponse.newBuilder()
                            .setRequestKey(requestKey) // 回传，用于识别对应的响应
                            .setAuthFactResponseJson(JSONUtil.toJsonStr(authFactResponse))
                            .build();
                    synchronized (responseObserver) {
                        responseObserver.onNext(rpcResponseObject);
                    }
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }

            @Override
            public void onError(Throwable t) {
                log.info("requestStreamObserver error");
                // t.printStackTrace();
                if (responseObserver instanceof ServerCallStreamObserver) {
                    ServerCallStreamObserver serverCallStreamObserver = ((ServerCallStreamObserver) responseObserver);
                    if (serverCallStreamObserver.isCancelled()) {
                        //client close the stream.
                    } else {
                        try {
                            serverCallStreamObserver.onCompleted();
                        } catch (Throwable throwable) {
                            //ignore
                        }
                    }
                }
            }

            @Override
            public void onCompleted() {
                log.info("requestStreamObserver onCompleted");
                if (responseObserver instanceof ServerCallStreamObserver) {
                    ServerCallStreamObserver serverCallStreamObserver = ((ServerCallStreamObserver) responseObserver);
                    if (serverCallStreamObserver.isCancelled()) {
                        //client close the stream.
                    } else {
                        try {
                            serverCallStreamObserver.onCompleted();
                        } catch (Throwable throwable) {
                            //ignore
                        }
                    }
                }
            }
        };
        log.info("biStreamAuthFact end...........");

        return requestStreamObserver;
    }

    public void pushAuthFactToAgent(AuthFactResponse authFactResponse) {
        SystemInfoDTO systemInfoDTO = authFactResponse.getSystemInfoDTO();
        List<Table> tablesList = authFactResponse.getTables();
        for(Table t : tablesList) {
            // 通过 authFactResponse 找出需要更新的 "多个" request
            Iterator<Map.Entry<String,AuthFactRequest>> requestIterator = requestMap.entrySet().iterator();
            List<String> keyList = requestMap.entrySet().stream().filter( entry ->{ // TODO
                if(authFactResponse.getSystemInfoDTO().getId().equalsIgnoreCase(entry.getValue().getSystemId())
                    && authFactResponse.getSystemInfoDTO().getMicroServiceId().equalsIgnoreCase(entry.getValue().getSourceSystemId())
                    && t.getTabName().equalsIgnoreCase(entry.getValue().getDb().getDatabase()))
                    return true;
                else
                    return false;
            }).map( entry ->{ return entry.getKey(); }).collect(Collectors.toList());
            HashSet<StreamObserver<RPCAuthFactResponse>> pushedSet = new HashSet<>();
            // 通过 request 找到要回传的 requestKey，通过 requestKey 找出 response，整体 response 消重
            for(String reqKey: keyList) {
                RPCAuthFactResponse rpcAuthFactResponse = RPCAuthFactResponse.newBuilder()
                        .setRequestKey(reqKey) // 回传 key，目前暂时没用
                        .setAuthFactResponseJson(JSONUtil.toJsonStr(authFactResponse))
                        .build();
                StreamObserver<RPCAuthFactResponse> responseSo = responseAuthFactMap.get(reqKey);
                if(!pushedSet.contains(responseSo)) {
                    if (!((ServerCallStreamObserver) responseSo).isCancelled()) {
                        synchronized (responseSo) {
                            responseSo.onNext(rpcAuthFactResponse);
                        }
                        pushedSet.add(responseSo);
                    } else {
                        responseAuthFactMap.remove(reqKey, responseSo);
                    }
                }
            }
        }
    }
}
